package com.david;

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;


/**
 * 同步
 * 观察者和被观察者在同一个线程中
 */
public class RxJavaTest {

    public static void main(String[] args) {
        //被观察者
        Observable<String> girl=Observable.create(emitter -> {//emitter 发射器、发射体
            //onNext 可以无限次调用
            System.out.println(Thread.currentThread().getName());
            emitter.onNext("1");
            Thread.sleep(1000);
            emitter.onNext("2");
            Thread.sleep(1000);
            emitter.onNext("3");
            Thread.sleep(1000);
            emitter.onNext("4");
            Thread.sleep(1000);
            emitter.onNext("5");
            Thread.sleep(1000);
            emitter.onComplete();
        });

        //观察者
        Observer<String> man=new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("onSubscribe"+d);
            }

            @Override
            public void onNext(String s) {
                System.out.println(Thread.currentThread().getName());
                System.out.println("onNext"+s);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError"+e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        };

        girl.subscribe(man);

    }

}
